package org.example.service;

import com.alibaba.cloud.stream.binder.rocketmq.RocketMQMessageChannelBinder;
import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQProducerProperties;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.example.GreetingsStreams;
import org.example.pojo.Greetings;
import org.springframework.cloud.stream.binder.*;
import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.*;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.util.MimeTypeUtils;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.time.Duration;

@Service
@Slf4j
public class GreetingsService {
//    @Resource
    private  GreetingsStreams greetingsStreams;
    @Resource
    private BinderFactory binderFactory;
    @Resource
    StreamBridge streamBridge;

    private static MessageChannel messageChannel = null;
    private static Binding BINDING = null;
    private static Boolean isStart = false;
    public void buildCosumer() {
        Binder binder = binderFactory.getBinder("rocketmq", RocketMQMessageChannelBinder.class);
        SubscribableChannel messageChannel = new DirectChannel();
//        KafkaProducerProperties kafkaProducerProperties = new KafkaProducerProperties();
        RocketMQConsumerProperties rocketMQProducerProperties = new RocketMQConsumerProperties();
        ExtendedConsumerProperties<RocketMQConsumerProperties> extendedConsumerProperties = new ExtendedConsumerProperties<>(rocketMQProducerProperties);

        rocketMQProducerProperties.setBroadcasting(true);
        binder.bindConsumer("userDefine","user-group",messageChannel,extendedConsumerProperties);
        messageChannel.subscribe(message -> {
            byte[] rs = (byte[]) message.getPayload();
            log.info("Rev greetings {}", new String(rs));
            message.getHeaders().forEach((key, value) -> System.out.println(key + ":" + value));
            /*
            RocketMQ自带的Header包括如下：

            rocketmq_QUEUE_ID:0
            rocketmq_TOPIC:userDefine
            rocketmq_FLAG:0
            rocketmq_RECONSUME_TIMES:0
            rocketmq_MESSAGE_ID:0200000156B418B4AAC211B2FEAE0000
            rocketmq_SYS_FLAG:0
            id:2089d6c8-984f-0399-c975-52b18a680a8d
            rocketmq_BORN_HOST:10.41.31.163
            contentType:application/json
            rocketmq_BORN_TIMESTAMP:1691116143278
            timestamp:1691116143382
            */
        });

    }

    public void changeBinder(){
        BINDING.unbind();
        messageChannel = new DirectChannel();
        Binder binder = binderFactory.getBinder("kafka", KafkaMessageChannelBinder.class);
//        KafkaProducerProperties kafkaProducerProperties = new KafkaProducerProperties();
        KafkaProducerProperties kafkaProducerProperties = new KafkaProducerProperties();
        ExtendedProducerProperties<KafkaProducerProperties> extendedProducerProperties = new ExtendedProducerProperties<>(kafkaProducerProperties);
        BINDING = binder.bindProducer("userDefine",messageChannel,extendedProducerProperties);
        isStart = true;
    }
    public void builderMsgChannel(){
        Binder binder = binderFactory.getBinder("rocketmq", RocketMQMessageChannelBinder.class);
        messageChannel = new DirectChannel();
//        KafkaProducerProperties kafkaProducerProperties = new KafkaProducerProperties();
        RocketMQProducerProperties rocketMQProducerProperties = new RocketMQProducerProperties();
        rocketMQProducerProperties.setGroup("userDefine");
        ExtendedProducerProperties<RocketMQProducerProperties> extendedProducerProperties = new ExtendedProducerProperties<>(rocketMQProducerProperties);
        BINDING = binder.bindProducer("userDefine",messageChannel,extendedProducerProperties);
        isStart = true;
    }
    public void sendGreeting(final Greetings greetings) {
        if(!isStart){
            builderMsgChannel();
        }
        messageChannel.send(MessageBuilder
                .withPayload(JSON.toJSONString(greetings).getBytes(StandardCharsets.UTF_8))
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .build());
        log.info("Sending greetings {}", greetings);

    }
    public void sendGreetingByStreamBridge(final Greetings greetings){
        log.info("Sending greetings {}", greetings);
        streamBridge.send("greetings-out",MessageBuilder
                .withPayload(JSON.toJSONString(greetings))
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .setHeader("test","test1")
                .build());
    }
    public void sendGreetingByMsgChannel(final Greetings greetings) {
        log.info("Sending greetings {}", greetings);
                MessageChannel messageChannel = greetingsStreams.outboundGreetings();
        messageChannel.send(MessageBuilder
                .withPayload(JSON.toJSONString(greetings))
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .build());
    }
}